package com.futao.techsharingmq.kafka.basic.consumer;

import com.futao.techsharingmq.kafka.basic.declare.Topics;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaHandler;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.stereotype.Component;

/**
 * @author futao <1185172056@qq.com> <https://github.com/FutaoSmile>
 * @date 2021/3/3
 */
@Slf4j
@Component
public class OrderConsumer {

    /**
     * 测试消费者群内消费者数量为3，topic的分区为4的情况
     *
     * @param consumerRecord
     */
    @KafkaHandler
    @KafkaListener(topics = Topics.ORDER_TOPIC, concurrency = "3", groupId = "sb-g", autoStartup = "true")
    public void orderConsumer(ConsumerRecord<?, ?> consumerRecord) {
        log.info("接收到消息:[{}]", consumerRecord);
    }

    /**
     * 测试当spring.kafka.consumer.auto-offset-reset为latest的效果
     *
     * @param consumerRecord
     */
    @KafkaHandler
    @KafkaListener(topics = Topics.ORDER_TOPIC, concurrency = "3", groupId = "sb-g-2", autoStartup = "true")
    public void orderConsumer2(ConsumerRecord<?, ?> consumerRecord) {
        log.info("接收到消息:[{}]", consumerRecord);
    }

    /**
     * 测试当spring.kafka.consumer.auto-offset-reset为earliest的效果
     *
     * @param consumerRecord
     */
    @KafkaHandler
    @KafkaListener(topics = Topics.ORDER_TOPIC, concurrency = "3", groupId = "sb-g-3", autoStartup = "true")
    // @SendTo
    public void orderConsumer3(ConsumerRecord<?, ?> consumerRecord) {
        log.info("接收到消息:[{}]", consumerRecord);
    }

    //    @KafkaHandler
    //    @KafkaListener(topics = Topics.ORDER_TOPIC)
    //    public void orderConsumer2(Order order) {
    //        log.info("接收到消息:[{}]", JSON.toJSONString(order));
    //    }
}
